package com.cn.remote;

import com.cn.common.BrokerEntity;
import com.cn.common.ClientType;
import com.cn.common.RemotingResponseWrapper;
import com.cn.file.AppendMessageResult;
import com.cn.file.DefaultMessageStore;
import com.cn.file.MessageStore;
import com.cn.msg.Message;
import com.cn.producer.SendResult;
import com.cn.producer.SendStatus;
import com.cn.util.SerializationUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

public class RemoteClient extends RemoteAbstract {

    private ClientType clientType;

    private EventLoopGroup worker;

    private Bootstrap bootstrap;

    private boolean ok = false;

    /**
     * 远程服务地址
     */
    private String remoteAddr;


    private String address;
    private String name;

    private Channel channel;

    private BrokerEntity brokerEntity;
    private DefaultMessageStore defaultMessageStore;

    public RemoteClient(String remoteAddr, ClientType clientType, BrokerEntity brokerEntity) {
        this.clientType = clientType;
        this.remoteAddr = remoteAddr;
        this.brokerEntity = brokerEntity;
        init();
    }

    public RemoteClient(String remoteAddr, ClientType clientType) {
        this.clientType = clientType;
        this.remoteAddr = remoteAddr;
        init();
    }

    public RemoteClient(String remoteAddr, ClientType clientType, BrokerEntity brokerEntity, DefaultMessageStore defaultMessageStore) {
        this.clientType = clientType;
        this.remoteAddr = remoteAddr;
        this.defaultMessageStore = defaultMessageStore;
        init();
    }


    public void init() {
        this.worker = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();
        this.bootstrap.group(worker);
        this.bootstrap.channel(NioSocketChannel.class);
        this.bootstrap.option(ChannelOption.TCP_NODELAY, true);
        this.bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
        this.bootstrap.handler(new RemoteClientHandler(this));
    }


    public void start() throws InterruptedException {
        if (this.bootstrap == null) throw new RuntimeException("初始化失败");
        String[] s = this.remoteAddr.split(":");
        ChannelFuture f = bootstrap.connect(s[0], Integer.valueOf(s[1])).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                if (channelFuture.isSuccess()) {
                    channel = channelFuture.channel();
                    setOk(true);
                } else {
                    setOk(false);
                }
            }
        });
    }

    public void stop() {
        if (this.worker != null) {
            this.worker.shutdownGracefully();
        }
    }

    public SendResult invokeSync(Message message, MessageStore messageStore) throws InterruptedException {
        if (this.channel != null && this.channel.isActive()) {
            try {
                long t = 5000;
                //等待服务端返回
                ResponseFuture responseFuture = new ResponseFuture(0, this.channel, t);
                //将请求封装到容器中，等待服务端的返回
                RemotingResponseWrapper.add(0, responseFuture);

                //向服务端发送消息
                this.channel.writeAndFlush(message).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
                        if (f.isSuccess()) {
                            responseFuture.setSendRequestOK(true);
                            return;
                        } else {
                            responseFuture.setSendRequestOK(false);
                        }
                        RemotingResponseWrapper.remove(0);
                        responseFuture.putResponse(null);
                    }
                });

                //等待服务端返回
                AppendMessageResult result = responseFuture.waitResponse(t);
                if (null == result) {
                    return new SendResult(null, SendStatus.FLUSH_DISK_TIMEOUT);
                }
                //解析服务端返回的消息
                switch (result.getStatus()) {
                    case PUT_OK:
                        return new SendResult(result.getMsgId(), SendStatus.SEND_OK);
                    case PROPERTIES_SIZE_EXCEEDED:
                        return new SendResult(null, SendStatus.SEND_FAIL);
                    default:
                        return new SendResult(null, SendStatus.SEND_FAIL);
                }
            } finally {
                RemotingResponseWrapper.remove(0);
            }
        }
        return new SendResult(null, SendStatus.FLUSH_DISK_TIMEOUT);
    }

    public void sendMsg(Message message) {

        if (this.channel != null && this.channel.isActive()) {
            this.channel.writeAndFlush(message).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture f) throws Exception {
//                    if (f.isSuccess()) {
//                        System.out.println("写入成功---");
//                    } else {
//                        System.out.println("写入失败+++");
//                    }
                }
            });
        } else {
            System.out.println("通道已经关闭" + this.channel.isActive() + "--" + this.channel.isOpen() + "====");
        }
    }

    public DefaultMessageStore getDefaultMessageStore() {
        return defaultMessageStore;
    }

    public String getName() {
        return name;
    }

    public Channel getChannel() {
        return channel;
    }

    public BrokerEntity getBrokerEntity() {
        return brokerEntity;
    }

    public ClientType getClientType() {
        return clientType;
    }

    public boolean isOk() {
        return ok;
    }

    public void setOk(boolean ok) {
        this.ok = ok;
    }

    public void print() {
        System.out.println("连接服务端失败");
    }
}
